package com.dituhui.tool.elastic;

import java.net.InetAddress;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;

import org.apache.log4j.Logger;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequestBuilder;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.elasticsearch.transport.client.PreBuiltTransportClient;

import com.dituhui.tool.io.Constants;
import com.dituhui.tool.util.StringUtil;

import net.sf.json.JSONArray;
import net.sf.json.JSONObject;

public class ElasticSearchAPI {
	private static final Logger LOGGER = Logger.getLogger(ElasticSearchAPI.class);
	// private static TransportClient tclient;
	// private String HOST_ = Constants.ES_HOST_;
	// 以host为key
	private static Map<String, TransportClient> transportClientMap = new HashMap<String, TransportClient>();
	// CLUSTER_NAME_暂时从配置文件读取唯一值
	private static final String CLUSTER_NAME = Constants.getValue("ES.CLUSTER.NAME", "config.properties");
	private static final int STATUS_OK = 200;
	private static final int STATUS_FAIL = 500;
	private static final int STATUS_EMPTY_DATA = 501;

	// private long TimeoutMillis = 10000;

	/**
	 * 构造函数
	 */
	public ElasticSearchAPI() {
		// intClient();
	}

	/***
	 * 初始化 TransportClient
	 * 
	 * @param host ES服务地址
	 * @return TransportClient
	 */
	public TransportClient getTransportClient(String host) {
		TransportClient transportClient = null;

		try {
			transportClient = transportClientMap.get(host);
			if (transportClient == null) {
				synchronized (transportClientMap) {
					// 设置集群名称
					// 自动嗅探整个集群的状态，把集群中其它机器的ip地址加到客户端中
					// ES 2.2.0
					// Settings settings = Settings.settingsBuilder().put("cluster.name", CLUSTER_NAME).put("tclient.transport.sniff", true).build();
					// ES 5.2.x
					// sniff=true代表自动把集群下的机器添加到列表中
					Settings settings = Settings.builder().put("cluster.name", CLUSTER_NAME).put("client.transport.sniff", true).build();
					LOGGER.info("cluster.name=" + CLUSTER_NAME);

					// ES 2.2.0
					// transportClient = TransportClient.builder().settings(settings).build();
					// ES 5.2.x
					transportClient = new PreBuiltTransportClient(settings);
					String[] nodes = host.split(StringUtil.COMMA);
					for (String node : nodes) {
						// 跳过为空的node（当开头、结尾有逗号或多个连续逗号时会出现空node）
						if (node.length() > 0) {
							String[] hostPort = node.split(StringUtil.COLON);
							transportClient
									.addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(hostPort[0]), Integer.parseInt(hostPort[1])));
						}
					}
				}
				transportClientMap.put(host, transportClient);
			}
		} catch (Exception e) {
			// TODO:改成日志
			e.printStackTrace();
		}

		return transportClient;
	}

	/***
	 * 简单的查询封装
	 * 
	 * @param paramsJObj josn格式的参数
	 * @param host ES服务地址
	 * @return SearchResponse
	 */
	public SearchResponse search(JSONObject paramsJObj, String host) {
		// TimeoutMillis为ES集群超时时间
		// return builderSearchRequest(params_jObj_,
		// host).execute().actionGet(TimeoutMillis);
		return builderSearchRequest(paramsJObj, host).execute().actionGet();
	}

	/**
	 * 根据参数拼接一个 SearchRequestBuilder
	 * 
	 * @param paramsJObj josn格式的参数
	 * @param host ES服务地址
	 * @return SearchRequestBuilder
	 */
	protected SearchRequestBuilder builderSearchRequest(JSONObject paramsJObj, String host) {
		// LOGGER.info("in builderSearchRequest");
		SearchRequestBuilder searchRequestBuilder = null;

		// set index
		String[] index = null;
		if (paramsJObj.containsKey("index")) {
			index = paramsJObj.getString("index").split(",");
		} else {
			// index不能为空
			return null;
		}
		TransportClient transportClient = getTransportClient(host);
		searchRequestBuilder = transportClient.prepareSearch(index);

		// set _type
		if (paramsJObj.containsKey("_type")) {
			searchRequestBuilder.setTypes(paramsJObj.getString("_type"));
		}

		// routing
		if (paramsJObj.containsKey("routing")) {
			searchRequestBuilder.setRouting(paramsJObj.getString("routing"));
		}

		// set searchtype
		if (paramsJObj.containsKey("searchtype")) {
			/*------------------------
			 DFS_QUERY_THEN_FETCH 0
			 QUERY_THEN_FETCH     1
			 DFS_QUERY_AND_FETCH  2
			 QUERY_AND_FETCH      3
			 SCAN                 4
			 COUNT                5
			 -------------------------*/
			searchRequestBuilder.setSearchType(SearchType.fromId(Byte.parseByte(paramsJObj.getString("searchtype"))));
		}

		// ES 5.2.x 要使用SearchSourceBuilder
		SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();

		// set query_params
		if (paramsJObj.containsKey("term")) {
			JSONObject paramObj = paramsJObj.getJSONObject("term");

			Iterator<String> it = paramObj.keys();
			if (paramObj.size() > 1) {
				// 多字段查询
				BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
				while (it.hasNext()) {
					String key = it.next().toString();
					String value = (String) paramObj.get(key);
					// LOGGER.info("key = " + key + ",value = " + value);
					if (value != null && value.startsWith("jq_")) {
						// "jq_"开头的使用精准查询
						value = value.substring(3);
						// LOGGER.info("(2)key = " + key + ",value = " + value);
						boolQueryBuilder.must(QueryBuilders.termQuery(key, value));
					} else if (value != null && value.startsWith("qz_")) {
						// "qz_"开头的使用前缀查询
						value = value.substring(3);
						// System.out.println("key = " + key + ",value = " + value);
						boolQueryBuilder.must(QueryBuilders.prefixQuery(key, value));
					} else {
						// 模糊查询
						boolQueryBuilder.must(QueryBuilders.matchPhrasePrefixQuery(key, value));
						// queryBuilder = QueryBuilders.matchPhrasePrefixQuery(key, value);
						// 支持中文查询，使用termQuery有问题。
						// TODO: matchPhraseQuery可能效率更高，待验证
					}
				}
				// System.out.println("boolQueryBuilder=" + boolQueryBuilder.toString());
				// ES 2.2.0
				// searchRequestBuilder.setQuery(boolQueryBuilder);
				// ES 5.2.x
				searchSourceBuilder.query(boolQueryBuilder);
			} else {
				// 单一段查询
				// 只循环一次
				while (it.hasNext()) {
					String key = it.next().toString();
					String value = (String) paramObj.get(key);
					QueryBuilder queryBuilder = QueryBuilders.matchPhrasePrefixQuery(key, value);
					// ES 2.2.0
					// searchRequestBuilder.setQuery(queryBuilder);
					// ES 5.2.x
					searchSourceBuilder.query(queryBuilder);
				}
			}
		} else {
			return null;
		}

		// 返回记录个数
		int querySize = 0;
		if (paramsJObj.containsKey("size")) {
			querySize = paramsJObj.getInt("size");
			searchRequestBuilder.setFrom(0).setSize(querySize);
		}

		// returnFields，返回字段
		if (paramsJObj.containsKey("returnFields")) {
			Object fields = paramsJObj.get("returnFields");
			// System.out.println("fields=" + fields);
			if (fields != null) {
				JSONArray jsonArray = (JSONArray) fields;
				if (jsonArray != null) {
					int size = jsonArray.size();
					String[] arr = new String[size];
					for (int i = 0; i < size; i++) {
						arr[i] = jsonArray.getString(i);
						// System.out.println("arr[" + i + "] = " + arr[i]);
					}

					// ES 5.2.x
					String[] includes = arr;
					String[] excludes = null;
					// a)指定单个字段
					// String include = "admincode";
					// searchSourceBuilder.fetchSource(include, null).from(0).size(11);
					// b)指定多个字段
					searchSourceBuilder.fetchSource(includes, excludes).from(0).size(querySize);
					// c)对象作为参数
					// fetchSource必须为true
					// boolean fetchSource = true;
					// FetchSourceContext fetchSourceContext = new FetchSourceContext(fetchSource, includes, excludes);
					// searchSourceBuilder.fetchSource(fetchSourceContext);
					searchRequestBuilder.setSource(searchSourceBuilder);
					// ES 2.2.0
					// searchRequestBuilder.addFields(arr);
				}
			}
		}

		// orderFields，排序字段
		if (paramsJObj.containsKey("orderFields")) {
			SortOrder order;
			if (paramsJObj.containsKey("sortOrder") && "asc".equalsIgnoreCase(paramsJObj.getString("sortOrder"))) {
				order = SortOrder.ASC;
			} else {
				order = SortOrder.DESC;
			}
			Object fields = paramsJObj.get("orderFields");
			// System.out.println("fields=" + fields);
			if (fields != null) {
				JSONArray jsonArray = (JSONArray) fields;
				if (jsonArray != null) {
					int size = jsonArray.size();
					for (int i = 0; i < size; i++) {
						String fieldName = jsonArray.getString(i);
						searchRequestBuilder.addSort(fieldName, order);
					}
				}
			}
		}

		// AbstractAggregationBuilder aggregation = AggregationBuilders.max("version").field(fieldName);
		// searchRequestBuilder.addAggregation(aggregation);

		// 设置是否按查询匹配度排序
		searchRequestBuilder.setExplain(true);

		return searchRequestBuilder;
	}

	/***
	 * 数据更新:批量更新 批量方式为ES的bulk方式
	 * 
	 * @param paramsJObj json对象
	 * @param host ES服务地址
	 * @return 更新结果
	 */
	public String update(JSONObject paramsJObj, String host) {
		TransportClient transportClient = getTransportClient(host);
		BulkRequestBuilder bulkRequest = transportClient.prepareBulk();

		JSONObject msgObj = new JSONObject();
		msgObj.accumulate("status", STATUS_FAIL);
		msgObj.accumulate("isSuccess", false);

		// _index 索引名称
		String index = null;
		if (paramsJObj.containsKey("index")) {
			index = paramsJObj.getString("index");
		} else {
			msgObj.accumulate("result", "key \"index\" is null");
			return msgObj.toString();
		} // if..else...

		String type = null;
		// _type 设置 _type标识
		if (paramsJObj.containsKey("_type")) {
			type = paramsJObj.getString("_type");
		} else {
			msgObj.accumulate("result", "key \"_type\" is null");
			return msgObj.toString();
		} // if..else...

		// routing
		String routing = null;
		if (paramsJObj.containsKey("routing")) {
			routing = paramsJObj.getString("routing");
		}

		// data object //文档数据
		if (paramsJObj.containsKey("datas")) {
			// 使用JSONArray，方便ES直接操作
			JSONArray datas = paramsJObj.getJSONArray("datas");

			int siz = datas.size();
			for (int i = 0; i < siz; i++) {
				JSONObject data = datas.getJSONObject(i);
				// tclient.prepareIndex(idx_,
				// type_).setSource(data).execute().actionGet();
				IndexRequestBuilder indexRequestBuilder = transportClient.prepareIndex(index, type).setSource(data);
				if (routing != null) {
					indexRequestBuilder.setRouting(routing);
				}
				// 使用bulk的批量方式进行操作
				bulkRequest.add(indexRequestBuilder);
			}

			// bulk统一进行execute
			BulkResponse response = bulkRequest.execute().actionGet();

			if (response.hasFailures()) {
				msgObj.accumulate("result", "processing failed. msg=" + response.buildFailureMessage());
			}

			msgObj.put("status", STATUS_OK);
			msgObj.put("isSuccess", true);
			msgObj.accumulate("result", "processing successed");

			return msgObj.toString();
		} else {
			msgObj.put("status", STATUS_EMPTY_DATA);
			msgObj.accumulate("result", "key \"datas\" is null");
			return msgObj.toString();
		} // if...else...

	}

	/***
	 * 删除索引
	 * 
	 * @param index 索引名称
	 * @param host host
	 */
	public void deleteIndex(String index, String host) {
		TransportClient transportClient = getTransportClient(host);
		DeleteIndexRequest delete = new DeleteIndexRequest(index);
		transportClient.admin().indices().delete(delete);
	}

}
